/*
 * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 *
 */

/*
 *
 *
 *
 *
 *
 * Written by Doug Lea, Bill Scherer, and Michael Scott with
 * assistance from members of JCP JSR-166 Expert Group and released to
 * the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
import java.util.*;
import java.util.Spliterator;
import java.util.Spliterators;

public class SynchronousQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    private static final long serialVersionUID = -3223113410248163686L;


    abstract static class Transferer<E> {
        /**
         *
         * @param e  可以为null，null的时候表示这个是一个request类型的请求，
         *           如果不是null，说明当前请求是一个data类型的请求
         * @param timed  true 表示指定了超时时间， false 表示不支持超时，直到匹配到为止
         * @param nanos 超时限制  单位：纳秒
         * @return  E 如果当前请求是一个request类型的请求
         *              返回值！=null 表示匹配成功
         *              返回值==null  表示超时或者被中断
         *          如果当前请求是data类型的请求，
         *              返回值！=null 表示匹配成功，返回当前线程put的数据
         *              返回值 ==null 表示data类型的请求超时 或者被中断
         */
        abstract E transfer(E e, boolean timed, long nanos);
    }

    /**
     * 表示获取当前系统所拥有的的cpu核心数
     */
    static final int NCPUS = Runtime.getRuntime().availableProcessors();
    /**
     *  表示指定了超时时间的话，最大的自旋次数
     *  为什么需要自旋操作？因为线程挂起唤醒站在cpu角度，是比较耗费资源的，
     *  涉及到用户态和内核太的切换浪费性能，自旋期间线程会一直检查自己的
     *  状态是否被匹配到，如果自旋期间被匹配到，直接返回，如果
     *  未被匹配到，达到某一指标后，还是会挂起。
     */
    static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;
    /**
     * 表示没有指定超时限制的时候，线程等待匹配时，自旋的次数。
     */
    static final int maxUntimedSpins = maxTimedSpins * 16;
    /**
     * 如果请求是指定超时限制的话，如果超时参数小于1000纳秒的时候，禁止挂起
     */
    static final long spinForTimeoutThreshold = 1000L;

    /**
     * 非公平模式实现的同步队列，内部数据结构是栈
     * @param <E>
     */
    static final class TransferStack<E> extends Transferer<E> {
        //表示node类型为请求类型
        static final int REQUEST    = 0;
        //表示node类型为数据类型
        static final int DATA       = 1;
        //表示node类型为匹配中类型
        //假设栈顶元素是request类型，当前请求为data的话，入栈会修改类型为FULFILLING
        //假设栈顶元素是data类型，当前请求为request的话，入栈会修改类型为FULFILLING
        static final int FULFILLING = 2;
        //判断当前模式是不是FULFILLING转态
        static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }

        static final class SNode {
            volatile SNode next;        //指向当前栈针的下一个栈针
            volatile SNode match;       //与当前node匹配的节点
            volatile Thread waiter;     //假设当前node线程自旋期间未匹配到，那么线程需要挂起，使用waiter
                                        //保存当前线程，方便匹配成功后，被唤醒。
            Object item;                //数据域，data不为空表示当前对应的请求类型为data类型，反之表示当前请求为request类型
            int mode;                   //表示当前node的模式


            SNode(Object item) {
                this.item = item;
            }
            //cas方式设置node对象的next字段
            boolean casNext(SNode cmp, SNode val) {
                //优化：为什么上来先判断？因为cas指令是原子指令，在平台执行的时候
                //同一时刻只能有一个cas指令执行
                //有了java层面的这一层判断，可以提升一部分性能。
                return cmp == next &&
                    UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
            }
            //尝试匹配
            //调用tryMatch对象时栈顶节点的下一个节点，与栈顶匹配的节点。
            //返回值：true表示匹配成功，否则匹配失败
            boolean tryMatch(SNode s) {
                //条件一：说明当前match还没被匹配过
                //条件二：cas设置match字段，表示当前node已经被匹配了。
                if (match == null &&
                    UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) {
                    //获取当前node的线程指针
                    Thread w = waiter;
                    //条件不为空，说明节点的线程已经挂起了
                    if (w != null) {
                        //释放当前节点对应的线程
                        waiter = null;
                        //唤醒node对应的线程
                        LockSupport.unpark(w);
                    }
                    return true;
                }
                return match == s;
            }
            //尝试取消
            void tryCancel() {
                //match字段保留当前node对象本身，表示这个node对象是取消状态，最终会被强制移除出栈
                UNSAFE.compareAndSwapObject(this, matchOffset, null, this);
            }
            //如果match保留的是当前node本身的引用，那表示当前node是取消状态
            boolean isCancelled() {
                return match == this;
            }

            private static final sun.misc.Unsafe UNSAFE;
            private static final long matchOffset;
            private static final long nextOffset;

            static {
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class<?> k = SNode.class;
                    matchOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("match"));
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }
        }
        //栈顶指针
        volatile SNode head;
        //cas设置栈顶元素
        boolean casHead(SNode h, SNode nh) {
            return h == head &&
                UNSAFE.compareAndSwapObject(this, headOffset, h, nh);
        }
        //s就是snode的引用
        static SNode snode(SNode s, Object e, SNode next, int mode) {
            //引用为空就创建对象
            if (s == null) s = new SNode(e);
            //设置他的模式
            s.mode = mode;
            //设置他的next域
            s.next = next;
            return s;
        }

        E transfer(E e, boolean timed, long nanos) {
            //s包装的是当前线程的node
            SNode s = null;
            //判断当前线程的模式 request ？ data
            int mode = (e == null) ? REQUEST : DATA;
            //自旋
            for (;;) {
                //获取栈顶元素
                SNode h = head;
                //h==null  栈为空
                //h.mode == mode  压栈操作，不匹配
                //case1：不管哪个成立，当前线程都需要入栈
                if (h == null || h.mode == mode) {
                    //设置了超时时间，并且超时时间小于等于0
                    //请求不支持阻塞等待
                    if (timed && nanos <= 0) {
                        //条件成立，说明栈顶已经是取消状态了，协助栈顶出栈
                        if (h != null && h.isCancelled())
                            casHead(h, h.next);
                        //大部分情况下，其实都只从这里直接走了
                        else
                            return null;
                    //当前栈顶为空 或者 模式与当前请求一致，且当前请求允许阻塞等待
                    } else if (casHead(h, s = snode(s, e, h, mode))) {//给当前请求cas设置成栈顶元素
                        //执行到这里，说明当前请求入栈成功
                        //在栈内等待被匹配
                        /*
                        * 正常情况，返回匹配的节点
                        * 取消情况：返回当前节点 s节点进去 返回s节点
                        * */
                        SNode m = awaitFulfill(s, timed, nanos);
                        /*条件成立：说明当前node是取消状态*/
                        if (m == s) {
                            /*将取消状态的节点出栈*/
                            clean(s);
                            /*最终返回null*/
                            return null;
                        }
                        /*条件1：说明栈顶有元素，此时就会来到条件2：栈顶的下一个节点就是当前节点
                          说明匹配的fulling 和当前 node还没出栈，需要协助出栈，将这两个节点出栈
                        * 执行到这里，说明当前node已经被匹配了
                        * */
                        if ((h = head) != null && h.next == s)
                            casHead(h, s.next);
                        return (E) ((mode == REQUEST) ? m.item : s.item);
                    }
                //case2：来到这里说明栈顶node的模式与当前请求的模式不一致
                } else if (!isFulfilling(h.mode)) {
                    if (h.isCancelled())
                        casHead(h, h.next);

                    else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
                        for (;;) {
                            SNode m = s.next;
                            if (m == null) {
                                casHead(s, null);
                                s = null;
                                break;
                            }
                            SNode mn = m.next;

                            if (m.tryMatch(s)) {
                                casHead(s, mn);
                                return (E) ((mode == REQUEST) ? m.item : s.item);
                            } else
                                s.casNext(m, mn);
                        }
                    }
                //case3：什么时候执行到case3？也就是栈顶模式==fulling模式，表示栈顶和栈顶下面的元素正在匹配..
                } else {
                    SNode m = h.next;
                    if (m == null)
                        casHead(h, null);
                    else {//帮助头节点匹配
                        SNode mn = m.next;
                        if (m.tryMatch(h))
                            casHead(h, mn);
                        else
                            h.casNext(m, mn);
                    }
                }
            }
        }

        SNode awaitFulfill(SNode s/*当前请求node*/, boolean timed/*是否支持超时*/, long nanos/*超时时长*/) {
            /*如果指定了超时限制，这里就是等待的截止时间，否则的话就是0*/
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            /*获取当前请求线程*/
            Thread w = Thread.currentThread();
            /*
                spins表示当前请求线程在下面的自旋中的自旋次数，
                如果达到spins时仍然没匹配成功，那么选择挂起当前线程
            */
            /*嵌套三元表达式*/
            int spins = (shouldSpin(s) ?
                         (timed ? maxTimedSpins : maxUntimedSpins) : 0);
            /*自旋检查逻辑*/
            for (;;) {
                if (w.isInterrupted())//当前线程被中断，需要设置node状态为取消状态
                    /*node对象的match指向当前node，说明当前node是取消状态*/
                    s.tryCancel();
                /*
                * 正常情况：有一个请求与当前node匹配成功，这个时候s.match 指向匹配节点
                * 取消情况：当前match指向当前node
                * */
                SNode m = s.match;//获取给定节点s的match节点
                if (m != null)//已经匹配到，返回匹配节点
                    /*可能是正常也可能是取消*/
                    return m;
                if (timed) {
                    //超时处理
                    nanos = deadline - System.nanoTime();
                    if (nanos <= 0L) {
                        s.tryCancel();//超时，取消s节点的匹配，match指向自身
                        continue;
                    }
                }
                /*条件成立：说明当前线程还可以进行自旋检查*/
                if (spins > 0)
                    //spins-1 自旋次数-1
                    spins = shouldSpin(s) ? (spins-1) : 0;
                else if (s.waiter == null)/*达到自旋次数，已经不允许进行自旋检查了*/
                    s.waiter = w; // establish waiter so can park next iter
                else if (!timed)//没有设定超时，直接阻塞
                    LockSupport.park(this);
                else if (nanos > spinForTimeoutThreshold)//阻塞指定超时时间
                    LockSupport.parkNanos(this, nanos);
            }
        }

        boolean shouldSpin(SNode s) {

            SNode h = head;
           /*
            * 1.当前线程是栈顶，允许自旋检查
            * 2.当前s节点自旋检查期间，又来了一个与s节点匹配的请求
            * 3. 执行到条件三，说明前面的两个条件都不成立，说明当前s不是栈顶元素，并且当前栈顶正在匹配中
                栈顶的下面的元素都允许自旋检查。
            */

            return (h == s || h == null || isFulfilling(h.mode));
        }

        void clean(SNode s) {
            s.item = null;   // forget item
            s.waiter = null; // forget thread

            SNode past = s.next;
            if (past != null && past.isCancelled())
                past = past.next;

            SNode p;
            while ((p = head) != null && p != past && p.isCancelled())
                casHead(p, p.next);

            while (p != null && p != past) {
                SNode n = p.next;
                if (n != null && n.isCancelled())
                    p.casNext(n, n.next);
                else
                    p = n;
            }
        }

        private static final sun.misc.Unsafe UNSAFE;
        private static final long headOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = TransferStack.class;
                headOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("head"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }

    static final class TransferQueue<E> extends Transferer<E> {

        static final class QNode {
            volatile QNode next;          // next node in queue
            volatile Object item;         // CAS'ed to or from null
            volatile Thread waiter;       // to control park/unpark
            final boolean isData;

            QNode(Object item, boolean isData) {
                this.item = item;
                this.isData = isData;
            }

            boolean casNext(QNode cmp, QNode val) {
                return next == cmp &&
                    UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
            }

            boolean casItem(Object cmp, Object val) {
                return item == cmp &&
                    UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
            }

            void tryCancel(Object cmp) {
                UNSAFE.compareAndSwapObject(this, itemOffset, cmp, this);
            }
            boolean isCancelled() {
                return item == this;
            }

            boolean isOffList() {
                return next == this;
            }

            private static final sun.misc.Unsafe UNSAFE;
            private static final long itemOffset;
            private static final long nextOffset;

            static {
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class<?> k = QNode.class;
                    itemOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("item"));
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }
        }

        transient volatile QNode head;
        transient volatile QNode tail;
        transient volatile QNode cleanMe;

        TransferQueue() {
            QNode h = new QNode(null, false); // initialize to dummy node.
            head = h;
            tail = h;
        }

        void advanceHead(QNode h, QNode nh) {
            if (h == head &&
                UNSAFE.compareAndSwapObject(this, headOffset, h, nh))
                h.next = h; // forget old next
        }

        void advanceTail(QNode t, QNode nt) {
            if (tail == t)
                UNSAFE.compareAndSwapObject(this, tailOffset, t, nt);
        }

        boolean casCleanMe(QNode cmp, QNode val) {
            return cleanMe == cmp &&
                UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
        }

        @SuppressWarnings("unchecked")
        E transfer(E e, boolean timed, long nanos) {

            QNode s = null; // constructed/reused as needed
            boolean isData = (e != null);//判断put or take

            for (;;) {
                QNode t = tail;
                QNode h = head;
                if (t == null || h == null)         // saw uninitialized value
                    continue;                       // spin

                if (h == t || t.isData == isData) { // empty or same-mode
                    QNode tn = t.next;
                    if (t != tail)                  // inconsistent read
                        continue;
                    if (tn != null) {               //尾节点滞后，更新尾节点 lagging tail
                        advanceTail(t, tn);
                        continue;
                    }
                    if (timed && nanos <= 0)        // can't wait
                        return null;
                    //为当前操作构造新节点，并放到队尾
                    if (s == null)
                        s = new QNode(e, isData);
                    if (!t.casNext(null, s))        // failed to link in
                        continue;

                    //推进tail
                    advanceTail(t, s);              // swing tail and wait
                    //等待匹配，并返回匹配节点的item，如果取消等待则返回该节点s
                    Object x = awaitFulfill(s, e, timed, nanos);
                    if (x == s) {                   // wait was cancelled
                        clean(t, s); //等待被取消，清除s节点
                        return null;
                    }

                    if (!s.isOffList()) {           // s节点尚未出列 not already unlinked
                        advanceHead(t, s);          // unlink if head
                        if (x != null)              // and forget fields
                            s.item = s;//item指向自身
                        s.waiter = null;
                    }
                    return (x != null) ? (E)x : e;

                    //take
                } else {                            // complementary-mode
                    QNode m = h.next;               // node to fulfill
                    if (t != tail || m == null || h != head)
                        continue;                   // inconsistent read

                    Object x = m.item;
                    if (isData == (x != null) ||    // m already fulfilled
                        x == m ||                   //m.item=m, m cancelled
                        !m.casItem(x, e)) {         // 匹配，CAS修改item为给定元素e lost CAS
                        advanceHead(h, m);          // 推进head，继续向后查找 dequeue and retry
                        continue;
                    }

                    advanceHead(h, m);              //匹配成功，head出列 successfully fulfilled
                    LockSupport.unpark(m.waiter);   //唤醒被匹配节点m的线程
                    return (x != null) ? (E)x : e;
                }
            }
        }

        Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
            /* Same idea as TransferStack.awaitFulfill */
            //同 TransferStack.awaitFulfill 采取策略一样
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            Thread w = Thread.currentThread();
            int spins = ((head.next == s) ?
                         (timed ? maxTimedSpins : maxUntimedSpins) : 0);
            for (;;) {
                if (w.isInterrupted())
                    s.tryCancel(e);//被中断，item引用指向自身
                Object x = s.item;
                if (x != e)
                    return x;
                if (timed) {
                    nanos = deadline - System.nanoTime();
                    if (nanos <= 0L) {
                        s.tryCancel(e);
                        continue;
                    }
                }
                if (spins > 0)
                    --spins;
                else if (s.waiter == null)
                    s.waiter = w;
                else if (!timed)
                    LockSupport.park(this);
                else if (nanos > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanos);
            }
        }

        void clean(QNode pred, QNode s) {
            s.waiter = null; // forget thread
            while (pred.next == s) { // Return early if already unlinked
                QNode h = head;
                QNode hn = h.next;   // Absorb cancelled first node as head
                //找到有效head节点
                if (hn != null && hn.isCancelled()) {
                    advanceHead(h, hn);
                    continue;
                }
                QNode t = tail;      // Ensure consistent read for tail
                if (t == h)//队列为空，直接返回
                    return;
                QNode tn = t.next;
                if (t != tail)//tail节点被其他线程修改，重新循环
                    continue;
                //找到tail节点
                if (tn != null) {
                    advanceTail(t, tn);
                    continue;
                }
                if (s != t) {        // If not tail, try to unsplice
                    QNode sn = s.next;
                    if (sn == s || pred.casNext(s, sn))//cas解除s的链接
                        return;
                }
                QNode dp = cleanMe;
                if (dp != null) {    // Try unlinking previous cancelled node
                    QNode d = dp.next;
                    QNode dn;
                    if (d == null ||               // d is gone or
                        d == dp ||                 // d is off list or
                        !d.isCancelled() ||        // d not cancelled or
                        (d != t &&                 // d not tail and
                         (dn = d.next) != null &&  //   has successor
                         dn != d &&                //   that is on list
                         dp.casNext(d, dn)))       // d unspliced
                        casCleanMe(dp, null);
                    if (dp == pred)
                        return;      // s is already saved node
                } else if (casCleanMe(null, pred))//原cleanMe为空，标记pred为cleanMe，延迟清除s节点
                    return;          // Postpone cleaning s
            }
        }

        private static final sun.misc.Unsafe UNSAFE;
        private static final long headOffset;
        private static final long tailOffset;
        private static final long cleanMeOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = TransferQueue.class;
                headOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("head"));
                tailOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("tail"));
                cleanMeOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("cleanMe"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }

    private transient volatile Transferer<E> transferer;

    public SynchronousQueue() {
        this(false);
    }

    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        if (transferer.transfer(e, false, 0) == null) {
            Thread.interrupted();
            throw new InterruptedException();
        }
    }

    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (e == null) throw new NullPointerException();
        if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
            return true;
        if (!Thread.interrupted())
            return false;
        throw new InterruptedException();
    }

    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        return transferer.transfer(e, true, 0) != null;
    }

    public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E e = transferer.transfer(null, true, unit.toNanos(timeout));
        if (e != null || !Thread.interrupted())
            return e;
        throw new InterruptedException();
    }

    public E poll() {
        return transferer.transfer(null, true, 0);
    }

    public boolean isEmpty() {
        return true;
    }

    public int size() {
        return 0;
    }

    public int remainingCapacity() {
        return 0;
    }

    public void clear() {
    }

    public boolean contains(Object o) {
        return false;
    }

    public boolean remove(Object o) {
        return false;
    }

    public boolean containsAll(Collection<?> c) {
        return c.isEmpty();
    }

    public boolean removeAll(Collection<?> c) {
        return false;
    }

    public boolean retainAll(Collection<?> c) {
        return false;
    }

    public E peek() {
        return null;
    }

    public Iterator<E> iterator() {
        return Collections.emptyIterator();
    }

    public Spliterator<E> spliterator() {
        return Spliterators.emptySpliterator();
    }

    public Object[] toArray() {
        return new Object[0];
    }

    public <T> T[] toArray(T[] a) {
        if (a.length > 0)
            a[0] = null;
        return a;
    }

    public int drainTo(Collection<? super E> c) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        int n = 0;
        for (E e; (e = poll()) != null;) {
            c.add(e);
            ++n;
        }
        return n;
    }

    public int drainTo(Collection<? super E> c, int maxElements) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        int n = 0;
        for (E e; n < maxElements && (e = poll()) != null;) {
            c.add(e);
            ++n;
        }
        return n;
    }


    static class WaitQueue implements java.io.Serializable { }
    static class LifoWaitQueue extends WaitQueue {
        private static final long serialVersionUID = -3633113410248163686L;
    }
    static class FifoWaitQueue extends WaitQueue {
        private static final long serialVersionUID = -3623113410248163686L;
    }
    private ReentrantLock qlock;
    private WaitQueue waitingProducers;
    private WaitQueue waitingConsumers;

    private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {
        boolean fair = transferer instanceof TransferQueue;
        if (fair) {
            qlock = new ReentrantLock(true);
            waitingProducers = new FifoWaitQueue();
            waitingConsumers = new FifoWaitQueue();
        }
        else {
            qlock = new ReentrantLock();
            waitingProducers = new LifoWaitQueue();
            waitingConsumers = new LifoWaitQueue();
        }
        s.defaultWriteObject();
    }

    private void readObject(java.io.ObjectInputStream s)
        throws java.io.IOException, ClassNotFoundException {
        s.defaultReadObject();
        if (waitingProducers instanceof FifoWaitQueue)
            transferer = new TransferQueue<E>();
        else
            transferer = new TransferStack<E>();
    }

    static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
                                  String field, Class<?> klazz) {
        try {
            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
        } catch (NoSuchFieldException e) {
            NoSuchFieldError error = new NoSuchFieldError(field);
            error.initCause(e);
            throw error;
        }
    }

}
